package cn.lsh.spark.pv;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

/**
 * pv：用户对网站的点击量（可重复）
 * uv：用户对网站的点击量（可重复，一个用户多次点击只算一次）
 */
public class PvSpark {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf();
		conf.setMaster("local").setAppName("pv");

		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> lines = sc.textFile("hdfs://mycluster/test/spark/pv/input/log.txt");
		System.out.println("*************PV*************");
		lines.mapToPair(line -> new Tuple2<>(line.split("\t")[5], 1)).reduceByKey(Integer::sum)
				.mapToPair(Tuple2::swap).sortByKey(false).mapToPair(Tuple2::swap).take(5).forEach(System.out::println);
		System.out.println("*************UV*************");
		lines.mapToPair(line -> new Tuple2<>(line.split("\t")[5], line.split("\t")[0])).distinct()
				.aggregateByKey(0, (t1, t2) -> t1 + 1, Integer::sum)
				.mapToPair(Tuple2::swap).sortByKey(false).mapToPair(Tuple2::swap).take(5).forEach(System.out::println);

		System.out.println("*************分区排序*************");
		lines.mapToPair(line -> new Tuple2<>(new Tuple2<>(line.split("\t")[5], line.split("\t")[1]), 1))
				.reduceByKey(Integer::sum)
				.mapToPair(t -> new Tuple2(new Tuple2<>(t._1()._1(), t._2()), t._1()._2()))
				//sortByKey自定义比较器必须实现Serializable接口，这里不能使用匿名内部类
				.sortByKey(new SerialComparator(), false).foreach(o -> System.out.println(o));

		/*JavaPairRDD<String, Iterable<Tuple2<Tuple2<String, String>, Integer>>> webGroupRdd = lines.mapToPair(line ->
				new Tuple2<>(new Tuple2<>(line.split("\t")[5], line.split("\t")[1]), 1))
				.reduceByKey(Integer::sum)
				.groupBy(t -> t._1()._1());
		JavaPairRDD<String, Tuple2<String, Integer>> stringTuple2JavaPairRDD = webGroupRdd.flatMapToPair(t -> {
			List<Tuple2<String, Tuple2<String, Integer>>> list = new ArrayList<>();
			t._2().forEach(o -> list.add(new Tuple2<>(t._1(), new Tuple2<>(o._1()._2(), o._2()))));
			return list.iterator();
		});
		webGroupRdd.mapToPair(Tuple2::swap).sortByKey(new Comparator<Iterable<Tuple2<Tuple2<String, String>, Integer>>>() {
			@Override
			public int compare(Iterable<Tuple2<Tuple2<String, String>, Integer>> o1, Iterable<Tuple2<Tuple2<String, String>, Integer>> o2) {
				return 0;
			}
		});*/
	}

	public static class SerialComparator implements Comparator<Tuple2<String, Integer>>, Serializable {

		@Override
		public int compare(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) {
			if (t1._1().equals(t2._1())) {
				return t1._2().compareTo(t2._2());
			}
			return t1._1().compareTo(t2._1());
		}
	}
}
